package com.atguigu.edu.realtime.dwd.db.app;

import com.atguigu.edu.realtime.common.base.BaseSQLApp;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeCartAdd extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdTradeCartAdd().start(
                10027,
                4,
                "dwd_transaction_add_cart"
        );
    }
    @Override
    public void handle1(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        readOdsDS(tableEnv,"dwd_transaction_add_cart");

        Table cartInfo = tableEnv.sqlQuery("select\n" +
                "    `data`['id'] id,\n" +
                "    `data`['user_id'] user_id,\n" +
                "    `data`['course_id'] course_id,\n" +
                "    `data`['course_name'] course_name,\n" +
                "    `data`['cart_price'] cart_price,\n" +
                "    `data`['create_time'] create_time,\n" +
                "    `data`['update_time'] update_time,\n" +
                "     ts   " +
                "from topic_db where `table`='cart_info'\n" +
                "and `database`='edu'\n" +
                "and (`type` = 'insert' or `type` = 'update')");

        //cartInfo.execute().print();

        tableEnv.executeSql("create table dwd_transaction_add_cart (\n" +
                "    id string,\n" +
                "    user_id string,\n" +
                "    course_id string,\n" +
                "    course_name string,\n" +
                "    cart_price string,\n" +
                "    create_time string,\n" +
                "    update_time string,\n" +
                "    ts bigint,\n" +
                "    primary key (id) not enforced\n" +
                " ) WITH (\n" +
                "  'connector' = 'upsert-kafka',\n" +
                "  'topic' = 'dwd_transaction_add_cart',\n" +
                "  'properties.bootstrap.servers' = 'hadoop104:9092',\n" +
                "  'key.format' = 'json',\n" +
                "  'value.format' = 'json'\n" +
                ");");

        cartInfo.executeInsert("dwd_transaction_add_cart");

    }
}
